/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.remoting.netty;

import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.Pair;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.common.SemaphoreReleaseOnlyOnce;
import org.apache.rocketmq.remoting.common.ServiceThread;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode;

import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;

public abstract class NettyRemotingAbstract {

    /**
     * Remoting logger instance.
     */
    private static final InternalLogger log = InternalLoggerFactory.getLogger(RemotingHelper.ROCKETMQ_REMOTING);

    /**
     * Semaphore to limit maximum number of on-going one-way requests, which protects system memory footprint.
     */
    protected final Semaphore semaphoreOneway;

    /**
     * Semaphore to limit maximum number of on-going asynchronous requests, which protects system memory footprint.
     */
    protected final Semaphore semaphoreAsync;

    /**
     * <pre>
     * 存放request已经发送,但是response尚未返回的请求.
     * key : 请求id  详见: {@link RemotingCommand#getOpaque()}
     * value : 响应Future
     *
     *
     * 1、请求正常返回处理:
     *          在{@link NettyRemotingClient.NettyClientHandler}中:
     *          当收到netty数据包之后,会判断 {@link RemotingCommand#getType()} 的类型,
     *          如果是response类型,调用 {@link NettyRemotingAbstract#processResponseCommand} 处理响应.
     *
     * 2、扫描超时的请求: 定时任务,定期调用 {@link NettyRemotingAbstract#scanResponseTable}方法:扫描过期的请求,然后去调用callback方法.
     * </pre>
     */
    protected final ConcurrentMap<Integer /* opaque */, ResponseFuture> responseTable = new ConcurrentHashMap<>(256);

    /**
     * <pre>
     * This container holds all processors per request code, aka, for each incoming request, we may look up the
     * responding processor in this map to handle the request.
     *
     * 这个容器保存每个请求代码的所有处理器，也就是，对于每个传入的请求，我们可以在这个映射中查找响应处理器来处理请求。
     * key : org.apache.rocketmq.common.protocol.RequestCode
     * value : 请求处理器,及对应的线程池
     * 备注:根据请求code找到对应的处理器,然后让处理器在指定的线程池中执行任务.那么这个任务的执行一定是异步的.todo:又是如何做到异步转同步呢?
     * </pre>
     */
    protected final HashMap<Integer/* request code */, Pair<NettyRequestProcessor, ExecutorService>> processorTable = new HashMap<>(64);

    /**
     * Executor to feed netty events to user defined {@link ChannelEventListener}.
     * netty事件执行器(就是个Runnable线程)
     */
    protected final NettyEventExecutor nettyEventExecutor = new NettyEventExecutor();

    /**
     * The default request processor to use in case there is no exact match in {@link #processorTable} per request code. <br/>
     * 默认的请求处理器.
     *
     * 解释:如果server端收到一个request,获取到请求的请求code,如果没有给这个请求code设置特定的请求处理器,则会使用此默认的请求处理器
     * 请求code详见: org.apache.rocketmq.common.protocol.RequestCode
     */
    protected Pair<NettyRequestProcessor, ExecutorService> defaultRequestProcessor;

    /**
     * SSL context via which to create {@link SslHandler}.
     */
    protected volatile SslContext sslContext;

    /**
     * custom rpc hooks
     */
    protected List<RPCHook> rpcHooks = new ArrayList<RPCHook>();


    static {
        NettyLogger.initNettyLogger();
    }

    /**
     * Constructor, specifying capacity of one-way and asynchronous semaphores.
     *
     * @param permitsOneway Number of permits for one-way requests.
     * @param permitsAsync Number of permits for asynchronous requests.
     */
    public NettyRemotingAbstract(final int permitsOneway, final int permitsAsync) {
        this.semaphoreOneway = new Semaphore(permitsOneway, true);
        this.semaphoreAsync = new Semaphore(permitsAsync, true);
    }

    /**
     * Custom channel event listener.
     *
     * @return custom channel event listener if defined; null otherwise.
     */
    public abstract ChannelEventListener getChannelEventListener();

    /**
     * Put a netty event to the executor.
     *
     * @param event Netty event instance.
     */
    public void putNettyEvent(final NettyEvent event) {
        this.nettyEventExecutor.putNettyEvent(event);
    }

    /**
     * Entry of incoming command processing.
     *
     * <p>
     * <strong>Note:</strong>
     * The incoming remoting command may be
     * <ul>
     * <li>An inquiry request from a remote peer component;</li>
     * <li>A response to a previous request issued by this very participant.</li>
     * </ul>
     * </p>
     *
     * 处理netty网络层'收到'的消息.<br/>
     * '收到'的消息,可能是请求,也有可能是响应.毕竟这是使用netty的tcp来实现的request和response.
     * tcp中是没有request和response的概念的.(http中是有的.)
     * 所以就需要中间件自己去实现.一般都是使用 [请求id] 来区分两个数据包是一个 请求&响应 组.
     *
     * 请求id 详见: {@link RemotingCommand#getOpaque}
     *
     * @param ctx Channel handler context.
     * @param msg incoming remoting command.
     * @throws Exception if there were any error while processing the incoming command.
     */
    public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
        final RemotingCommand cmd = msg;
        if (cmd != null) {
            switch (cmd.getType()) {
                case REQUEST_COMMAND:      // request   接收到的这个消息,是请求类型
                    processRequestCommand(ctx, cmd);
                    break;
                case RESPONSE_COMMAND:     // response  接收到的这个消息,是响应类型
                    processResponseCommand(ctx, cmd);
                    break;
                default:
                    break;
            }
        }
    }

    /**
     * 在请求处理前执行的钩子函数
     *
     * @param addr
     * @param request
     */
    protected void doBeforeRpcHooks(String addr, RemotingCommand request) {
        if (rpcHooks.size() > 0) {
            for (RPCHook rpcHook: rpcHooks) {
                rpcHook.doBeforeRequest(addr, request);
            }
        }
    }

    /**
     * 请求处理后钩子函数
     *
     * @param addr     地址
     * @param request  请求
     * @param response 响应
     */
    protected void doAfterRpcHooks(String addr, RemotingCommand request, RemotingCommand response) {
        if (rpcHooks.size() > 0) {
            for (RPCHook rpcHook: rpcHooks) {
                rpcHook.doAfterResponse(addr, request, response);
            }
        }
    }


    /**
     * Process incoming request command issued by remote peer.
     * 处理网络io接收的request命令.
     *
     * 注意:这个方法是[处理收到的request类型的请求]!!! 并不是发送request请求.
     * 发送request详见: {@link #invokeAsyncImpl} {@link #invokeSyncImpl} {@link #invokeOnewayImpl}
     *
     * @param ctx channel handler context.
     * @param cmd request command.
     */
    public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        // 获取这种请求类型,指定的请求处理器及线程池.如果没有指定,就使用默认的请求处理器及线程池.
        Pair<NettyRequestProcessor, ExecutorService> pair = this.processorTable.getOrDefault(cmd.getCode(), this.defaultRequestProcessor);
        // 请求id
        final int opaque = cmd.getOpaque();

        if (pair != null) {
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    try {
                        // 在处理接收到的request之前的钩子.
                        doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
                        //
                        final RemotingResponseCallback callback = new RemotingResponseCallback() {
                            @Override
                            public void callback(RemotingCommand response) {
                                // 处理request之后的回调
                                doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
                                // 如果请求不是oneway通信,则需要给对方发送response响应数据.
                                if (!cmd.isOnewayRPC()) {
                                    // 如果response不为null,才会给client端发送数据.
                                    // 这种情况举个栗子:consumer从broker拉取消息,如果PULL_MESSAGE请求带有hasSuspendFlag,则broker会把consumer的这个拉取请求保存起来,
                                    //                 (有单独的异步线程监听新mq消息,当有新消息时,给consumer返回数据.)
                                    //                 而当前处理PULL_MESSAGE请求的线程,设置 response = null; 此时broker就不会给consumer发送数据,并且处理当前也会释放.
                                    //                 详见: PullRequest pullRequest = new PullRequest(request, channel, pollingTimeMills, now, offset, subscriptionData, messageFilter);
                                    //                       this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);
                                    //                       response = null;
                                    if (response != null) {
                                        // 设置请求id
                                        response.setOpaque(opaque);
                                        // 标记当前发送的数据包是response类型的
                                        response.markResponseType();
                                        try {
                                            // 使用netty网络通道,把response消息发送给对方.
                                            ctx.writeAndFlush(response);
                                        } catch (Throwable e) {
                                            log.error("process request over, but response failed", e);
                                            log.error(cmd.toString());
                                            log.error(response.toString());
                                        }
                                    }
                                }
                            }
                        };
                        // 异步处理nett请求
                        if (pair.getObject1() instanceof AsyncNettyRequestProcessor) {
                            AsyncNettyRequestProcessor processor = (AsyncNettyRequestProcessor)pair.getObject1();
                            processor.asyncProcessRequest(ctx, cmd, callback);
                        } else {
                            NettyRequestProcessor processor = pair.getObject1();
                            // 同步处理接收到的request
                            RemotingCommand response = processor.processRequest(ctx, cmd);
                            // 同步处理request之后,调用callback
                            callback.callback(response);
                        }
                    } catch (Throwable e) {
                        log.error("process request exception", e);
                        log.error(cmd.toString());

                        if (!cmd.isOnewayRPC()) {
                            final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR,
                                RemotingHelper.exceptionSimpleDesc(e));
                            response.setOpaque(opaque);
                            ctx.writeAndFlush(response);
                        }
                    }
                }
            };

            // 如果需要拒绝此请求.则返回response:  SYSTEM_BUSY 系统忙
            if (pair.getObject1().rejectRequest()) {
                final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY, "[REJECTREQUEST]system busy, start flow control for a while");
                response.setOpaque(opaque);
                ctx.writeAndFlush(response);
                return;
            }

            try {
                // 搞成一个异步任务,交给线程池去执行
                final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
                ExecutorService executorService = pair.getObject2();
                // 放到线程池中.
                executorService.submit(requestTask);
            } catch (RejectedExecutionException e) {
                if ((System.currentTimeMillis() % 10000) == 0) {
                    log.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel())
                        + ", too many requests and system thread pool busy, RejectedExecutionException "
                        + pair.getObject2().toString()
                        + " request code: " + cmd.getCode());
                }

                // 异常时,如果不是oneway的rpc请求,则返回response
                if (!cmd.isOnewayRPC()) {
                    final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY, "[OVERLOAD]system busy, start flow control for a while");
                    response.setOpaque(opaque);
                    ctx.writeAndFlush(response);
                }
            }
        } else {
            // 没有找到当前请求code对应的请求处理器,所以返回:REQUEST_CODE_NOT_SUPPORTED 不支持请求代码
            String error = " request type " + cmd.getCode() + " not supported";
            final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
            response.setOpaque(opaque);
            ctx.writeAndFlush(response);
            log.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
        }
    }

    /**
     * Process response from remote peer to the previous issued requests.
     * 处理网络io接收的response命令.
     *
     * @param ctx channel handler context.
     * @param cmd response command instance. 响应实例
     */
    public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
        // 请求id requestId  详见 : RemotingCommand.getOpaque
        final int opaque = cmd.getOpaque();
        // 获取response
        final ResponseFuture responseFuture = responseTable.get(opaque);
        if (responseFuture != null) {
            // 设置response,但是不countDown
            responseFuture.setResponseCommand(cmd);
            // 从集合中移除此请求.
            responseTable.remove(opaque); // netty收到response响应数据,从map中移除请求future去处理.

            // callback不为空,则去调用callback  [这种情况说明client是异步请求]
            if (responseFuture.getInvokeCallback() != null) {
                executeInvokeCallback(responseFuture);
            } else {
                // 否则设置响应                 [这种情况说明client是同步请求]
                responseFuture.putResponse(cmd); // 设置response并countDown.那么发送request的线程就能跳出阻塞.
                responseFuture.release(); // 释放资源.
            }
        } else {
            log.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
            log.warn(cmd.toString());
        }
    }

    /**
     * 执行callback <br/>
     *
     * Execute callback in callback executor. If callback executor is null, run directly in current thread <br/>
     * 在回调执行器中执行回调。如果回调执行器为空，则直接在当前线程中运行
     */
    private void executeInvokeCallback(final ResponseFuture responseFuture) {
        boolean runInThisThread = false; // 是否在当前线程执行callback
        // callback方法的执行线程池
        ExecutorService executor = this.getCallbackExecutor();
        if (executor != null) {
            try {
                // 如果存在线程池,则使用对应的线程池去执行callback
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            // 执行callback
                            responseFuture.executeInvokeCallback();
                        } catch (Throwable e) {
                            log.warn("execute callback in executor exception, and callback throw", e);
                        } finally {
                            // 释放信号量.   限流器释放一个资源
                            responseFuture.release();
                        }
                    }
                });
            } catch (Exception e) {
                // 如果向异步线程池提交任务时异常了,则还在当前线程执行callback.(异常的原因可能是执行器太忙了——maybe executor busy)
                runInThisThread = true;
                log.warn("execute callback in executor exception, maybe executor busy", e);
            }
        } else {
            // 未指定callback在特定线程池中执行,则:callback在当前线程执行
            runInThisThread = true;
        }

        if (runInThisThread) {
            try {
                // 执行callback
                responseFuture.executeInvokeCallback();
            } catch (Throwable e) {
                log.warn("executeInvokeCallback Exception", e);
            } finally {
                // 释放资源.   释放线程器.
                responseFuture.release();
            }
        }
    }



    /**
     * Custom RPC hook.
     * Just be compatible with the previous version, use getRPCHooks instead.
     */
    @Deprecated
    protected RPCHook getRPCHook() {
        if (rpcHooks.size() > 0) {
            return rpcHooks.get(0);
        }
        return null;
    }

    /**
     * Custom RPC hooks.
     *
     * @return RPC hooks if specified; null otherwise.
     */
    public List<RPCHook> getRPCHooks() {
        return rpcHooks;
    }


    /**
     * This method specifies thread pool to use while invoking callback methods.
     *
     * @return Dedicated thread pool instance if specified; or null if the callback is supposed to be executed in the
     * netty event-loop thread.
     */
    public abstract ExecutorService getCallbackExecutor();

    /**
     * <p>
     * This method is periodically invoked to scan and expire deprecated request.
     * 定期调用此方法来扫描和过期已弃用的请求。
     * 
     * 备注:这个方法在client和server端的timer中都调用了:都需要去扫描responseTable请求表.
     * 原因:因为server端也可以给client消息.
     * (并不是基于http协议的,http协议只有client给server端发请求,但是RocketMQ使用netty框架基于tcp协议,
     * {@link RemotingCommand} 可以双向发送:client可以给server发请求,server也可以给client发送请求.
     * todo:比如:(以下指示猜想,还没证实.先这么理解)
     *  consumer在使用 MQPushConsumer 这种'推送'模式消费消息时,broker需要把消息主动推送给consumer,此时可能会用到server给client发送消息.
     * </p>
     */
    public void scanResponseTable() {
        final List<ResponseFuture> timeoutResponseFutureList = new LinkedList<ResponseFuture>();
        Iterator<Entry<Integer, ResponseFuture>> it = this.responseTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<Integer, ResponseFuture> next = it.next();
            ResponseFuture rep = next.getValue();

            // 判断是否超时,如果超时了,就释放资源并移除此超时的请求.
            if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
                rep.release();
                it.remove();
                // 超时的请求
                timeoutResponseFutureList.add(rep);
                log.warn("remove timeout request, " + rep);
            }
        }

        // 对已经timeout的请求,调用callback方法
        for (ResponseFuture timeoutFuture : timeoutResponseFutureList) {
            try {
                executeInvokeCallback(timeoutFuture);
            } catch (Throwable e) {
                log.warn("scanResponseTable, operationComplete Exception", e);
            }
        }
    }

    /**
     * 同步调用
     *
     * @param channel       netty网络通道.使用此网络通道包请求序列化之后,通过tcp协议发送到server端
     * @param request       请求
     * @param timeoutMillis 超时时间
     * @return
     * @throws InterruptedException
     * @throws RemotingSendRequestException
     * @throws RemotingTimeoutException
     */
    public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis)
            throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
        // requestId
        final int opaque = request.getOpaque();

        try {
            final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
            this.responseTable.put(opaque, responseFuture);
            final SocketAddress addr = channel.remoteAddress();
            // 发送请求,并设置监听.
            channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture f) throws Exception {
                    // 请求发送完毕,如果成功了,就设置请求发送ok.否则就设置请求发送不ok
                    if (f.isSuccess()) {
                        responseFuture.setSendRequestOK(true);
                    } else {
                        responseFuture.setSendRequestOK(false);
                        responseTable.remove(opaque); // 通过netty发送req失败,移除此请求
                        responseFuture.setCause(f.cause()); // 设置不ok的原因
                        responseFuture.putResponse(null); // 设置返回的响应体为null
                        log.warn("send a request command to channel <" + addr + "> failed.");
                    }
                }
            });

            // [等待响应response]
            // note:这里就是异步转同步的一般实现方案.就是直接在发送完请求之后,调用获取结果的接口.
            //  这样的话,就会阻塞到这里,知道请求返回.其实也就是 responseFuture.putResponse() 方法被调用时,才会解除阻塞.
            RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
            // 如果响应数据为空,
            if (null == responseCommand) {
                // 如果请求发送成功——说明是远程响应超时
                if (responseFuture.isSendRequestOK()) {
                    throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis, responseFuture.getCause());
                } else {
                    // 请求没有发送成功——抛出异常:请求发送异常
                    throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
                }
            }

            return responseCommand;
        } finally {
            this.responseTable.remove(opaque); // 同步发送req,处理完,在finally中,移除请求.
        }
    }

    /**
     * 异步发送请求
     *
     * @param channel        netty网络通道.使用此网络通道包请求序列化之后,通过tcp协议发送到server端
     * @param request        请求
     * @param timeoutMillis  超时时间
     * @param invokeCallback 发送请求之后的回调函数
     * @throws InterruptedException
     * @throws RemotingTooMuchRequestException
     * @throws RemotingTimeoutException
     * @throws RemotingSendRequestException
     */
    public void invokeAsyncImpl(final Channel channel, final RemotingCommand request,
                                final long timeoutMillis, final InvokeCallback invokeCallback)
            throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        long beginStartTime = System.currentTimeMillis();
        // 当前请求的请求id
        final int opaque = request.getOpaque();
        boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) {
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);
            // 获取信号量所用时间.
            long costTime = System.currentTimeMillis() - beginStartTime;
            if (timeoutMillis < costTime) {
                // 获取信号量超时,则释放信号量,并且报错异常:超时了!
                once.release();
                throw new RemotingTimeoutException("invokeAsyncImpl call timeout");
            }

            // 响应Future.
            final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis - costTime, invokeCallback, once);
            // 根据请求id保存到map中.
            // note:这里放到map中的作用:当client收到response消息时,根据请求id,从这个map中获取对应的future,并调用回调方法.
            //  所以从源码阅读技巧方面,应该去看 responseTable.get 方法在哪使用的.
            this.responseTable.put(opaque, responseFuture);
            try {
                // 使用netty,通过tpc发送请求.并添加tpc发送状态的监听器
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception {
                        // 请求tcp消息发送成功:
                        if (f.isSuccess()) {
                            // 设置 发送请求已经ok了.
                            responseFuture.setSendRequestOK(true);
                            return;
                        }
                        // 如果发送请求不成功,则设置这个请求id失败了.
                        requestFail(opaque);
                        log.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
                    }
                });
            } catch (Exception e) {
                responseFuture.release();
                log.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
                throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
            }
        } else { // 未申请到信号量,发送request太多导致的.
            if (timeoutMillis <= 0) {// 未设置超时时间
                throw new RemotingTooMuchRequestException("invokeAsyncImpl invoke too fast");
            } else {// 设置超时时间:获取信号量超时.
                String info =
                    String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                        timeoutMillis,
                        this.semaphoreAsync.getQueueLength(),
                        this.semaphoreAsync.availablePermits()
                    );
                log.warn(info);
                throw new RemotingTimeoutException(info);
            }
        }
    }

    /**
     * 根据请求id,设置请求tcp消息时,发送失败.
     *
     * @param opaque 请求id {@link RemotingCommand#getOpaque()}
     */
    private void requestFail(final int opaque) {
        // 从响应map中,根据请求id移除.
        ResponseFuture responseFuture = responseTable.remove(opaque);
        if (responseFuture != null) {
            // 设置发送请求不ok.
            responseFuture.setSendRequestOK(false);
            // 设置响应消息体为null
            responseFuture.putResponse(null);
            try {
                // 执行callback
                executeInvokeCallback(responseFuture);
            } catch (Throwable e) {
                log.warn("execute callback in requestFail, and callback throw", e);
            } finally {
                // 释放资源.具体就是:释放信号量.  (这个信号量感觉就是个限流器的作用)
                responseFuture.release();
            }
        }
    }

    /**
     * mark the request of the specified channel as fail and to invoke fail callback immediately
     * 将指定通道的请求标记为失败并立即调用失败回调
     * @param channel the channel which is close already 已经关闭的通道
     */
    protected void failFast(final Channel channel) {
        for (Entry<Integer, ResponseFuture> entry : responseTable.entrySet()) {
            if (entry.getValue().getProcessChannel() == channel) {
                // 请求id
                Integer opaque = entry.getKey();
                if (opaque != null) {
                    requestFail(opaque);
                }
            }
        }
    }

    /**
     * oneway单向请求发送 <br/>
     * (oneway:表示客户端发向服务端发送请求,并不关心服务端的响应是成功还是失败,只要请求发送成功就行了.)
     *
     * @param channel netty网络通道.使用此网络通道包请求序列化之后,通过tcp协议发送到server端
     * @param request 请求信息
     * @param timeoutMillis 请求发送超时时间(是用来获取信号量是否超时的)
     * @throws InterruptedException
     * @throws RemotingTooMuchRequestException
     * @throws RemotingTimeoutException
     * @throws RemotingSendRequestException
     */
    public void invokeOnewayImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis)
        throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        // 标记当前请求是单向 RPC
        request.markOnewayRPC();
        // 申请信号量.作用:限制并发量.
        boolean acquired = this.semaphoreOneway.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) {
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreOneway);
            try {
                // 使用netty网络channel发送请求.并且在发送完毕后,释放信号量.
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception {
                        // 操作完成时,释放信号量.
                        once.release();
                        if (!f.isSuccess()) {
                            log.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
                        }
                    }
                });
            } catch (Exception e) {
                // 发送请求异常,则也释放信号量,并报错
                once.release();
                log.warn("write send a request command to channel <" + channel.remoteAddress() + "> failed.");
                // 报错:远程发送请求异常.
                throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
            }
        } else {// 未获取到信号量.可能是当前处理oneway的请求太多了
            if (timeoutMillis <= 0) {// 未设置超时时间,直接报错:远程处理太多的请求.
                throw new RemotingTooMuchRequestException("invokeOnewayImpl invoke too fast");
            } else {// 设置了超时时间:获取信号量超时timeout.
                String info = String.format(
                    "invokeOnewayImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreOnewayValue: %d",
                    timeoutMillis,
                    this.semaphoreOneway.getQueueLength(),
                    this.semaphoreOneway.availablePermits()
                );
                log.warn(info);
                throw new RemotingTimeoutException(info);
            }
        }
    }

    class NettyEventExecutor extends ServiceThread {
        private final LinkedBlockingQueue<NettyEvent> eventQueue = new LinkedBlockingQueue<NettyEvent>();
        private final int maxSize = 10000;

        /**
         * 添加事件
         */
        public void putNettyEvent(final NettyEvent event) {
            if (this.eventQueue.size() <= maxSize) {
                this.eventQueue.add(event);
            } else {
                // 如果超过最大长度,则打印日志并丢弃.
                log.warn("event queue size[{}] enough, so drop this event {}", this.eventQueue.size(), event.toString());
            }
        }

        @Override
        public void run() {
            log.info(this.getServiceName() + " service started");

            // 获取所有的监听器
            final ChannelEventListener listener = NettyRemotingAbstract.this.getChannelEventListener();

            while (!this.isStopped()) {
                try {
                    // 从队列中拿到'事件',交给监听器去处理.
                    NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
                    if (event != null && listener != null) {
                        switch (event.getType()) {
                            case IDLE:
                                listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CLOSE:
                                listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CONNECT:
                                listener.onChannelConnect(event.getRemoteAddr(), event.getChannel());
                                break;
                            case EXCEPTION:
                                listener.onChannelException(event.getRemoteAddr(), event.getChannel());
                                break;
                            default:
                                break;

                        }
                    }
                } catch (Exception e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }

            log.info(this.getServiceName() + " service end");
        }

        @Override
        public String getServiceName() {
            return NettyEventExecutor.class.getSimpleName();
        }
    }
}
